package me.xmaples.flink

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object DataSetWordCount {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment

    env.readTextFile(exmapleTextFile)
      .flatMap { ln => ln.split(raw"\W+").filter(_.nonEmpty).map(_.toLowerCase) }
      .map((_, 1))
      .groupBy(0)
      .sum(1)
      .print()

    //对于DataSet，调用其.print()/.collect()等之后调用ExecutionEnvironment.execute()将会报错，因前者会间接调用.execute()

    //env.execute(getClass.getSimpleName)

    /*上述统计结果等同以下不用flink的代码的统计结果
    val wc2 =
      io.Source.fromFile(exmapleTextFile)
        .getLines()
        .flatMap { ln => ln.split(raw"\W+").filter(_.nonEmpty).map(_.toLowerCase) }
        .toList
        .groupBy(w => w)
        .mapValues(_.size)*/

    println("main done")
  }
}

object StreamingWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.readTextFile(exmapleTextFile)
      .flatMap { ln => ln.split(raw"\W+").filter(_.nonEmpty).map(_.toLowerCase) }
      .map((_, 1))
      .keyBy(0)
      .sum(1)
      .print()

    env.execute()

    println("main done")
  }
}

object TextStreamTumblingCountedWindowWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.readTextFile(exmapleTextFile)
      .flatMap { ln => ln.split(raw"\W+").filter(_.nonEmpty).map(_.toLowerCase()) }
      .map((_, 1))
      .keyBy(0)
      //每个单词（键）中每出现5次即形成一个计数窗口(其大小为5个元素，即打印出来的也应该是5)，
      // 统计值（实际上都不用统计就能估测出是5）并打印。出现次数少于5的单词不会被打印（因统计频率要求为每5个）。
      .countWindow(5)
      .sum(1)
      .print()

    env.execute()
  }
}

object TextStreamSlidingCountedWindowWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.readTextFile(exmapleTextFile)
      .flatMap { ln => ln.split(raw"\W+").filter(_.nonEmpty).map(_.toLowerCase()) }
      .map((_, 1))
      .keyBy(0)
      //每个单词（键）中每出现3次即形成一个计数窗口(其大小为5个元素，即打印出来的也应该是5，每个键的第一个窗口的统计值应该是3)，
      // 统计值（每个键的第一个窗口统计值为3，后续窗口统计值为5）并打印。出现次数少于3的单词不会被打印（因统计频率要求为每3个）。
      .countWindow(5, 3)
      .sum(1)
      .print()

    env.execute()
  }
}

object RangeStreamTumblingCountedWindowNumCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.fromCollection(1 to 100)
      .map { n => (n % 3, 1) } //流数据：1,2,0依次周期出现33轮（共计99个数字）的流
      .keyBy(0)
      //每20个统计一次，对键化流。因此要求每个键中每20个打印一次窗口内的统计值，
      //而每个窗口大小是20个元素，因此打印出来的计数值都是20，（每个键中）最后13个元素因达不到20个(窗口大小参数)的要求而被丢弃不计算不打印
      .countWindow(20)
      .sum(1)
      .print()

    /*输出
    1 -> 20
    2 -> 20
    0 -> 20
    后不再输出，程序退出，仅有60个元素实际参数计算，后面每个键13个（共计39个）元素因不满足窗口大小而不能形成窗口进而参与计算
    */

    env.execute()
  }
}

object TimestampedStreamSlidingTimedWindow {
  def main(args: Array[String]): Unit = {
    import scala.language.postfixOps

    val tmBaseMs = java.time.Instant.now().toEpochMilli
    //自当前时间起每秒一个事件，事件的值是一个数值，其代表时间差（秒级）对3取模后的值（即范围在[0,2]）
    //亦即，自上述时间起，依次从字母a,b,c中每秒取一个出来作为一个事件
    val timestampedNumSeq = (0 to 59).map { n => (n % 3 + 'a' toChar, tmBaseMs + n * 1000) } //0,1,2转为字母 a,b,c

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //设置时间度量种类为事件时间。默认是处理时间

    env.fromCollection(timestampedNumSeq)
      .assignAscendingTimestamps(_._2) //告诉引擎如何取出事件时间
      .map { kn => (kn._1, 1) }
      .keyBy(0)
      /*
       参数size控制窗口大小，在这里也就决定了聚合出的统计值。
       窗口大小20秒，数据流中20秒时长内一个字母出现的次数为20/3约等于6.7，也就说统计值不超过7（输出中字母对应的值不超过7）
        */
      /*
       参数slide控制频率，也就决定了生成多少个窗口（即打印出时同个键，也就这里的字母a或b或c，的统计次数）
       slide=15，事件时间区间总长约60秒（每个键的起止略有不同，但悬殊不超过一个窗口时长），60/15=4，加上区间边界(以及水印的生成)可能使得多出1个，也就说窗口数量为4个或5个
       （即打印的输出中，一个字母有4个或5个统计值）
        */
      .timeWindow(Time.seconds(20), Time.seconds(15))
      .sum(1)
      .print()

    env.execute(getClass.getSimpleName)
  }
}

object SocketStreamTumblingTimedWindowWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.socketTextStream("127.0.0.1", 17315)
      .flatMap(_.split(raw"\s"))
      .filter(_.nonEmpty)
      .map { w => (w.toLowerCase, 1) }
      .keyBy(0)
      //每10秒进行一次统计（窗口）
      .timeWindow(Time.seconds(10))
      .sum(1)
      .print()

    env.execute(getClass.getSimpleName)
  }
}

object SocketStreamSlidingTimedWindowWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env
      .socketTextStream("127.0.0.1", 17315)
      .flatMap(_.split(raw"\s"))
      .filter(_.nonEmpty)
      .map { w => (w.toLowerCase, 1) }
      .keyBy(0)
      //每6秒触发一次统计，统计10秒时长内的数据
      .timeWindow(Time.seconds(20), Time.seconds(12))
      .sum(1)
      .print()

    env.execute(getClass.getSimpleName)
  }
}

